package com.edu.yx.app.App_04_DwsApp.Dws_06_Play;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.edu.yx.app.APP_01_BaseApp.BaseApp;
import com.edu.yx.bean.PlayVideoAggBean;
import com.edu.yx.bean.PlayVideoBean;
import com.edu.yx.util.TempUtil;
import com.edu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;

import static com.edu.yx.common.Constant.*;

public class Dws_01_Play_ChapterCount extends BaseApp {

    public static void main(String[] args) {
        new Dws_01_Play_ChapterCount().init(
                4061,
                2,
                "Dws_01_Play_ChapterCount",
                TOPIC_DWD_PLAY_VIDEO_DETAIL
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //1.粒度是一行一次播放行为（30秒生成一次），需要将同一个时间段的所有播放行为进行合并
        //todo 如果不开窗，想要实现多数据进行聚合，选择使用定时器
        SingleOutputStreamOperator<PlayVideoBean> parseToPojoStream = parseToPojo(stream);

        //2.开窗聚合
        /**
         *  PlayVideoBean(stt=, edt=, chapterId=22917, videoId=2022, uId=91, curDate=, play_sec=286, ts=1662112785333)
         *  PlayVideoBean(stt=, edt=, chapterId=22917, videoId=2022, uId=92, curDate=, play_sec=491, ts=1662112785484)
         */
        SingleOutputStreamOperator<PlayVideoAggBean> windowAndAggStream = windowAndAgg(parseToPojoStream);

        //3.写入doris,这个不计算，直接在sugar算
        writeToDoris(windowAndAggStream);



    }

    private void writeToDoris(SingleOutputStreamOperator<PlayVideoAggBean> windowAndAggStream) {

        windowAndAggStream
                .map(bean ->{

                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String json = JSON.toJSONString(bean, config);

                    return json;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_play_chapter_count"));
    }

    private SingleOutputStreamOperator<PlayVideoAggBean> windowAndAgg(SingleOutputStreamOperator<PlayVideoBean> parseToPojoStream) {
        //聚合之前要先重新封装一个pojo

        return parseToPojoStream
                .map(bean -> {

                    String chapterId = bean.getChapterId();
                    String uId = bean.getUId();
                    Long play_sec = bean.getPlay_sec();

                    return new PlayVideoAggBean(
                            "",
                            "",
                            chapterId,
                            "",
                            new HashSet<>(Collections.singleton(uId)),
                            0L,
                            1L,
                            play_sec,
                            bean.getTs()
                    );

                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<PlayVideoAggBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(bean -> bean.getChapterId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<PlayVideoAggBean>() {
                            @Override
                            public PlayVideoAggBean reduce(PlayVideoAggBean value1,
                                                           PlayVideoAggBean value2) throws Exception {

                                value1.setVideoCt(value1.getVideoCt() + value2.getVideoCt());
                                value1.setPlay_sec(value1.getPlay_sec()+value2.getPlay_sec());
                                value1.getUIdSet().addAll(value2.getUIdSet());

                                return value1;
                            }
                        }, new ProcessWindowFunction<PlayVideoAggBean, PlayVideoAggBean, String, TimeWindow>() {
                            @Override
                            public void process(String s,
                                                Context context,
                                                Iterable<PlayVideoAggBean> elements,
                                                Collector<PlayVideoAggBean> out) throws Exception {
                                PlayVideoAggBean bean = elements.iterator().next();

                                bean.setStt(TempUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(TempUtil.toDateTime(context.window().getEnd()));
                                bean.setCurDate(TempUtil.toDate(System.currentTimeMillis()));

                                bean.setUIdCt((long) bean.getUIdSet().size());

                                out.collect(bean);

                            }
                        }
                );
    }

    private SingleOutputStreamOperator<PlayVideoBean> parseToPojo(DataStreamSource<String> stream) {

        return stream
                .map(strjson -> {
                    JSONObject jsonObject = JSON.parseObject(strjson);

                    String video_id = jsonObject.getString("video_id");
                    String uid = jsonObject.getString("uid");
                    Long play_sec = jsonObject.getLong("play_sec");
                    Long ts = jsonObject.getLong("ts");
                    String chapter_id = jsonObject.getString("chapter_id");

                    return new PlayVideoBean(
                            "",
                            "",
                            chapter_id,
                            video_id,
                            uid,
                            "",
                            play_sec,
                            ts
                    );

                })
                //将视频、uid、时间合并为一个key，将这些数据进行聚合，成一条数据
                .keyBy(bean -> bean.getVideoId() + "_" + bean.getUId() + "_" + bean.getTs())

                .process(new KeyedProcessFunction<String, PlayVideoBean, PlayVideoBean>() {

                    private ValueState<PlayVideoBean> sumPlaySec;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //设立一个单值，记录播放时间playSec，后续进来的都跟他相加
                        sumPlaySec = getRuntimeContext().getState(new ValueStateDescriptor<PlayVideoBean>("sumPlaySec", PlayVideoBean.class));

                    }

                    @Override
                    public void processElement(PlayVideoBean value,
                                               Context ctx,
                                               Collector<PlayVideoBean> out) throws Exception {

                        if (sumPlaySec.value() == null) {

                            //更新值，并且注册定时器，
                            // todo 由于播放时间累加的次数不确定，但是不能把每次的结果都抛出去，设置定时器往外抛
                            //  如果不用定时器，这个地方没窗口，来一个数据抛一个，显然不合理，用定时实现类似窗口的画地为牢的功能
                            sumPlaySec.update(value);
                            ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 3000);

                        } else {

                            //数据不是第一次进来，进行加法
                            Long stateSum = sumPlaySec.value().getPlay_sec();
                            Long nowSum = stateSum + value.getPlay_sec();

                            //创建一个对象，更新到状态内
                            PlayVideoBean playVideoBean = new PlayVideoBean(
                                    "",
                                    "",
                                    value.getChapterId(),
                                    value.getVideoId(),
                                    value.getUId(),
                                    "",
                                    nowSum,
                                    value.getTs()
                            );

                            sumPlaySec.update(playVideoBean);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp,
                                        OnTimerContext ctx,
                                        Collector<PlayVideoBean> out) throws Exception {
                        out.collect(sumPlaySec.value());
                    }
                });


    }
}
